package com.leyantech.ecp.agenx.common.sender;

import com.leyantech.chaos.kafka.KafkaProducerStringClient;
import com.leyantech.utility.config.ConfigUtil;

import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.name.Named;
import io.vertx.core.Future;
import io.vertx.core.impl.VertxInternal;
import io.vertx.core.impl.future.PromiseInternal;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Objects;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-24.
 */
@Slf4j
@Singleton
public class KafkaSender implements Sender<String> {

  private final VertxInternal vertx;

  private final KafkaProducerStringClient producer;

  private final Supplier<String> messageTopic;

  @Inject
  public KafkaSender(VertxInternal vertx,
      KafkaProducerStringClient producer,
      @Named("topicKey") String topicKey) {
    final Supplier<String> messageTopic = ConfigUtil.getDynamicConfig(
        topicKey,
        Function.identity(),
        ""
    );
    Preconditions.checkArgument(StringUtils.isNotBlank(messageTopic.get()), "topic is blank.");
    this.vertx = Objects.requireNonNull(vertx, "vertx is null");
    this.producer = Objects.requireNonNull(producer, "producer is null");
    this.messageTopic = messageTopic;
  }

  @Override
  public Future<Void> send(String address, String message) {
    final PromiseInternal<Void> promise = vertx.promise();
    final String topic = messageTopic.get();
    producer.send(topic, address, message, null, (metadata, exception) -> {
      if (exception != null) {
        log.error("send failed, {} {} {} ", topic, address, message, exception);
        promise.fail(exception);
      } else {
        log.debug("send succeed, {} {} {}", topic, address, message);
        promise.complete();
      }
    });
    return promise.future();
  }

}
